#include "Avalon/IO/PipedMessageReader.hpp"
#include "Avalon/Base/Algorithm.hpp"
#include "Avalon/IO/BufferReader.hpp"
#include "Avalon/IO/EndOfFileException.hpp"
#include "Avalon/IO/PipedMessageWriter.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Threading;
using namespace boost;
using namespace std;

PipedMessageReader::PipedMessageReader()
    : m_writeMutex(new mutex()),
      m_result(NULL),
      m_buffer(NULL),
      m_destination(NULL),
      m_writer(NULL) {}

PipedMessageReader::~PipedMessageReader() {
  Break();
  Clear(m_messages);
}

void PipedMessageReader::Break() {
  unique_lock<mutex> writeLock(*m_writeMutex);
  unique_lock<mutex> readLock(m_readMutex);
  if(m_writer == NULL) {
    return;
  }
  m_writer->InternalBreak();
  InternalBreak(readLock, writeLock);
}

void PipedMessageReader::Break(const exception_ptr& cause) {
  unique_lock<mutex> writeLock(*m_writeMutex);
  unique_lock<mutex> readLock(m_readMutex);
  m_cause = cause;
  if(m_writer == NULL) {
    return;
  }
  m_writer->InternalBreak();
  InternalBreak(readLock, writeLock);
}

void PipedMessageReader::Read(void* destination, int size,
    Out<Async<int> > sizeRead) {
  unique_lock<mutex> writeLock(*m_writeMutex);
  unique_lock<mutex> readLock(m_readMutex);
  while(!m_messages.empty()) {
    BufferReader* topMessage = m_messages.front();
    Async<int> readResult;
    topMessage->Read(destination, size, Store(readResult));
    if(readResult.GetState() != Async<int>::EXCEPTION) {
      readLock.unlock();
      writeLock.unlock();
      sizeRead->SetResult(readResult.Get());
      return;
    }
    delete topMessage;
    m_messages.pop_front();
  }
  if(m_writer == NULL) {
    readLock.unlock();
    writeLock.unlock();
    if(m_cause == NULL) {
      sizeRead->SetException(IOException("Pipe broken."));
    } else {
      sizeRead->SetException(m_cause);
    }
    return;
  }
  m_result = sizeRead.Get();
  m_buffer = NULL;
  m_destination = destination;
  m_size = size;
}

void PipedMessageReader::Read(Buffer* destination, int size,
    Out<Async<int> > sizeRead) {
  unique_lock<mutex> writeLock(*m_writeMutex);
  unique_lock<mutex> readLock(m_readMutex);
  while(!m_messages.empty()) {
    BufferReader* topMessage = m_messages.front();
    Async<int> readResult;
    topMessage->Read(destination, size, Store(readResult));
    if(readResult.GetState() != Async<int>::EXCEPTION) {
      readLock.unlock();
      writeLock.unlock();
      sizeRead->SetResult(readResult.Get());
      return;
    }
    delete topMessage;
    m_messages.pop_front();
  }
  if(m_writer == NULL) {
    readLock.unlock();
    writeLock.unlock();
    if(m_cause == NULL) {
      sizeRead->SetException(IOException("Pipe broken."));
    } else {
      sizeRead->SetException(m_cause);
    }
    return;
  }
  m_result = sizeRead.Get();
  m_buffer = destination;
  m_destination = NULL;
  m_size = size;
}

void PipedMessageReader::Connect(PipedMessageWriter* writer) {
  m_writer = writer;
}

void PipedMessageReader::InternalBreak(unique_lock<mutex>& readLock,
    unique_lock<mutex>& writeLock) {
  m_writer = NULL;
  if(m_result == NULL) {
    return;
  }
  Async<int>* result = m_result;
  m_result = NULL;
  m_buffer = NULL;
  m_destination = NULL;
  readLock.unlock();
  writeLock.unlock();
  if(m_cause == NULL) {
    result->SetException(IOException("Pipe broken."));
  } else {
    result->SetException(m_cause);
  }
}

void PipedMessageReader::Write(const Buffer& data,
    unique_lock<mutex>& readLock, unique_lock<mutex>& writeLock) {
  BufferReader* message = new BufferReader(data);
  if(m_result == NULL) {
    m_messages.push_back(message);
    return;
  }
  Async<int> readResult;
  if(m_buffer != NULL) {
    message->Read(m_buffer, m_size, Store(readResult));
  } else {
    message->Read(m_destination, m_size, Store(readResult));
  }
  if(readResult.Get() < data.GetSize()) {
    m_messages.push_back(message);
  } else {
    delete message;
  }
  Async<int>* result = m_result;
  m_result = NULL;
  m_buffer = NULL;
  m_destination = NULL;
  readLock.unlock();
  writeLock.unlock();
  result->SetResult(readResult.Get());
}
